package study.bigdata.flink.job.mysql;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.Event;
import com.github.shyiko.mysql.binlog.event.EventHeaderV4;
import com.github.shyiko.mysql.binlog.event.EventType;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.io.IOException;

/**
 * https://blog.csdn.net/lgq2016/article/details/126533174
 */
@Slf4j
public class MySqlBinlogSourceFunction extends RichSourceFunction<String> {

    private volatile boolean running = true;

    private BinaryLogClient client;

    @Override
    public void open(Configuration parameters) throws Exception {
        //
        log.info(">>>>>> open open open");
        //
        super.open(parameters);
    }

    @Override
    public void close() throws Exception {
        log.info(">>>>>> close close close");
        //
        client.disconnect();
        //
        super.close();
    }

    @Override
    public void cancel() {
        running = false;
    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        log.info(">>>>>> run run run");
        //
        client = new BinaryLogClient("127.0.0.1", 3306, "root", "root");
        // 连接
        client.setConnectTimeout(2000);
        // 保活
        client.setKeepAlive(true);
        client.setKeepAliveInterval(1000000L);
        client.setHeartbeatInterval((long) (1000000L * 0.8f));

        // 事件反序列化
        EventDeserializer deserializer = new EventDeserializer();
        deserializer.setCompatibilityMode(EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG);
        client.setEventDeserializer(deserializer);
        // 事件监听
        client.registerEventListener(event -> parse(ctx, event));
        //
        client.setBinlogFilename("");
        client.setBinlogPosition(11);
        //
        while (running) {
            log.info("1111111111111111");
            try {
                client.connect();
            } catch (IOException ex) {
                log.error("{}:{}监听器错误");
            }
        }
    }

    private void parse(SourceContext<String> sourceContext, Event event) {
        //
        EventHeaderV4 header = event.getHeader();
        EventType eventType = header.getEventType();
        System.out.println(">>>>>> " + eventType.toString());
        sourceContext.collect("ssssssssssssss");

        switch (eventType) {
            case TABLE_MAP:
                break;
            case EXT_UPDATE_ROWS:
                log.info("{}", event.getData());
                break;
            case EXT_WRITE_ROWS:
                log.info("");
                break;
            case EXT_DELETE_ROWS:
                break;
            default:
                break;
        }
    }
}
